package drds.binlog.parse.mysql;

import drds.binlog.binlog_event.LogEvent;
import drds.binlog.common.Authors;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.common.position.LogPosition;
import drds.binlog.common.utils.JsonUtils;
import drds.binlog.data_object.Entry;
import drds.binlog.data_object.EntryType;
import drds.binlog.database_driver.packets.server.FieldPacket;
import drds.binlog.database_driver.packets.server.ResultSetPacket;
import drds.binlog.event_filter.EventFilter;
import drds.binlog.event_filter.aviater.RegexFilter;
import drds.binlog.parse.*;
import drds.binlog.parse.exception.ParseException;
import drds.binlog.parse.mysql.dbsync.BinlogLogEventConverter;
import drds.binlog.parse.mysql.dbsync.TableMetaDataCache;
import drds.binlog.parse.mysql.table_meta_data.BaseOnDatabaseTableMetaDataGetter;
import drds.binlog.parse.mysql.table_meta_data.DefaultTableMetaDataGetterFactory;
import drds.binlog.parse.mysql.table_meta_data.TableMetaDataGetter;
import drds.binlog.parse.mysql.table_meta_data.TableMetaDataGetterFactory;
import drds.common.$;
import drds.common.Author;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.io.IOException;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 基于向mysql server复制binlog实现
 * <p>
 * <pre>
 * 1. 自身不控制mysql主备切换，由ha机制来控制. 比如接入tddl/cobar/自身心跳包成功率
 * 2. 切换机制
 * </pre>
 */
@Slf4j
@Author(name = Authors.LI_YANG)
public class EventParser extends AbstractEventParser implements IEventParser
{
    protected static final long BINLOG_START_OFFEST = 4L;

    private final AtomicLong eventsPublishBlockingTime = new AtomicLong(0L);
    // instance received binlog bytes
    private final AtomicLong receivedBinlogBytes = new AtomicLong(0L);
    protected TableMetaDataGetterFactory tableMetaDataGetterFactory = new DefaultTableMetaDataGetterFactory();
    @Setter
    @Getter
    protected boolean enableTableMetaData = false;
    @Setter
    @Getter
    protected TableMetaDataGetter tableMetaDataGetter;
    // 编码信息
    @Setter
    @Getter
    protected byte connectionCharsetNumber = (byte) 33;
    @Setter
    @Getter
    protected Charset connectionCharset = Charset.forName("UTF-8");
    @Setter
    @Getter
    protected boolean filterQueryDcl = false;
    @Setter
    @Getter
    protected boolean filterQueryDml = false;
    @Setter
    @Getter
    protected boolean filterQueryDdl = false;
    @Setter
    @Getter
    protected boolean filterRows = false;
    @Setter
    @Getter
    protected boolean filterTableError = false;


    // binlog信息
    @Setter
    @Getter
    protected EntryPosition entryPosition;


    @Setter
    @Getter
    private int defaultConnectionTimeoutInSeconds = 30;                // sotimeout


    ////////////////////////
    @Setter
    @Getter
    private int receiveBufferSize = 64 * 1024;
    @Setter
    @Getter
    private int sendBufferSize = 64 * 1024;

    // 心跳检查信息
    @Setter
    @Getter
    private String detectingSQL;                                          // 心跳sql
    @Setter
    @Getter
    private Connection connection;                                        // 查询meta信息的链接
    @Setter
    @Getter
    private TableMetaDataCache tableMetaDataCache;                                        // 对应meta
    @Setter
    @Getter
    private int fallbackIntervalInSeconds = 60;                // 切换回退时间
    @Setter
    @Getter
    private BinlogFormat[] binlogFormats;                                  // 支持的binlogFormat,如果设置会执行强校验
    @Setter
    @Getter
    private BinlogImage[] binlogImages;                                   // 支持的binlogImage,如果设置会执行强校验
    //特殊异常处理参数
    @Setter
    @Getter
    private int dumpErrorCount = 0;                 // binlogDump失败异常计数
    @Setter
    @Getter
    private int dumpErrorCountThreshold = 2;                 // binlogDump失败异常计数阀值

    protected BinlogParser buildParser()
    {
        BinlogLogEventConverter binlogLogEventConverter = new BinlogLogEventConverter();
        if (schemaNameAndTableNameRegexFilter != null && schemaNameAndTableNameRegexFilter instanceof RegexFilter)
        {
            binlogLogEventConverter.setSchemaNameAndTableNameRegexFilter((RegexFilter) schemaNameAndTableNameRegexFilter);
        }

        if (blackSchemaNameAndTableNameRegexFilter != null && blackSchemaNameAndTableNameRegexFilter instanceof RegexFilter)
        {
            binlogLogEventConverter.setBlackSchemaNameAndTableNameRegexFilter((RegexFilter) blackSchemaNameAndTableNameRegexFilter);
        }

        binlogLogEventConverter.setConnectionCharset(connectionCharset);
        binlogLogEventConverter.setFilterQueryDcl(filterQueryDcl);
        binlogLogEventConverter.setFilterQueryDml(filterQueryDml);
        binlogLogEventConverter.setFilterQueryDdl(filterQueryDdl);
        binlogLogEventConverter.setFilterRows(filterRows);
        binlogLogEventConverter.setFilterTableError(filterTableError);
        return binlogLogEventConverter;
    }

    public void setSchemaNameAndTableNameRegexFilter(EventFilter schemaNameAndTableNameRegexFilter)
    {
        super.setSchemaNameAndTableNameRegexFilter(schemaNameAndTableNameRegexFilter);

        // 触发一下filter变更
        if (schemaNameAndTableNameRegexFilter != null && schemaNameAndTableNameRegexFilter instanceof RegexFilter && binlogParser instanceof BinlogLogEventConverter)
        {
            ((BinlogLogEventConverter) binlogParser).setSchemaNameAndTableNameRegexFilter((RegexFilter) schemaNameAndTableNameRegexFilter);
        }
    }

    /**
     * 回滚到指定位点
     */
    protected boolean processTableMetaData(EntryPosition entryPosition)
    {
        if (tableMetaDataGetter != null)
        {
            if (entryPosition.getTimestamp() == null || entryPosition.getTimestamp() <= 0)
            {
                throw new ParseException("use gtid and TableMetaData TSDB should be config timestamp > 0");
            }

            return tableMetaDataGetter.rollback(entryPosition);
        }

        return true;
    }


    public void start() throws ParseException
    {


        if (enableTableMetaData)
        {
            if (tableMetaDataGetter == null)
            {
                synchronized (IEventParser.class)
                {
                    try
                    {
                        // 设置当前正在加载的通道，加载spring查找文件时会用到该变量
                        System.setProperty("canal.instance.destination", destination);
                        // 初始化
                        tableMetaDataGetter = tableMetaDataGetterFactory.build(destination);
                    } finally
                    {
                        System.setProperty("canal.instance.destination", "");
                    }
                }
            }
        }

        super.start();
    }

    public void stop() throws ParseException
    {
        if (connection != null)
        {
            try
            {
                connection.disconnect();
            } catch (IOException e)
            {
                log.error("ERROR # disconnect metadata connection for address:{}", connection.getConnector()
                        .getSocketAddress(), e);
            }
        }

        if (tableMetaDataCache != null)
        {
            tableMetaDataCache.clearTableMeta();
        }
        if (enableTableMetaData)
        {
            tableMetaDataGetterFactory.destory(destination);
            tableMetaDataGetter = null;
        }

        super.stop();
    }

    public void setBlackSchemaNameAndTableNameRegexFilter(EventFilter eventFilter)
    {
        super.setBlackSchemaNameAndTableNameRegexFilter(eventFilter);

        // 触发一下filter变更
        if (eventFilter != null && eventFilter instanceof RegexFilter
                && binlogParser instanceof BinlogLogEventConverter)
        {
            ((BinlogLogEventConverter) binlogParser).setBlackSchemaNameAndTableNameRegexFilter((RegexFilter) eventFilter);
        }
    }

    protected IMultiStageCoprocessor buildMultiStageCoprocessor()
    {
        MultiStageCoprocessor multiStageCoprocessor = new MultiStageCoprocessor(parallelBufferSize,
                parallelThreadSize,
                (BinlogLogEventConverter) binlogParser,
                entryListTransaction,
                destination);
        multiStageCoprocessor.setEventsPublishBlockingTime(eventsPublishBlockingTime);
        return multiStageCoprocessor;
    }

    protected IConnection buildConnection()
    {
        return buildConnection(this.authenticationInfo);
    }

    protected void preDump(IConnection connection)
    {
        if (!(connection instanceof Connection))
        {
            throw new ParseException("Unsupported connection logType : " + connection.getClass().getSimpleName());
        }

        if (binlogParser != null && binlogParser instanceof BinlogLogEventConverter)
        {
            this.connection = (Connection) connection.fork();
            try
            {
                this.connection.connect();
            } catch (IOException e)
            {
                throw new ParseException(e);
            }

            if (binlogFormats != null && binlogFormats.length > 0)
            {
                BinlogFormat binlogFormat = (this.connection).getBinlogFormat();
                boolean found = false;
                for (BinlogFormat supportFormat : binlogFormats)
                {
                    if (supportFormat != null && binlogFormat == supportFormat)
                    {
                        found = true;
                        break;
                    }
                }

                if (!found)
                {
                    throw new ParseException("Unsupported BinlogFormat " + binlogFormat);
                }
            }

            if (binlogImages != null && binlogImages.length > 0)
            {
                BinlogImage image = (this.connection).getBinlogImage();
                boolean found = false;
                for (BinlogImage supportImage : binlogImages)
                {
                    if (supportImage != null && image == supportImage)
                    {
                        found = true;
                        break;
                    }
                }

                if (!found)
                {
                    throw new ParseException("Unsupported BinlogImage " + image);
                }
            }

            if (tableMetaDataGetter != null && tableMetaDataGetter instanceof BaseOnDatabaseTableMetaDataGetter)
            {
                ((BaseOnDatabaseTableMetaDataGetter) tableMetaDataGetter).setConnection(this.connection);
                ((BaseOnDatabaseTableMetaDataGetter) tableMetaDataGetter).setEventFilter(schemaNameAndTableNameRegexFilter);
                ((BaseOnDatabaseTableMetaDataGetter) tableMetaDataGetter).setBlackFilter(blackSchemaNameAndTableNameRegexFilter);
            }

            tableMetaDataCache = new TableMetaDataCache(this.connection, tableMetaDataGetter);
            ((BinlogLogEventConverter) binlogParser).setTableMetaDataCache(tableMetaDataCache);
        }
    }

    protected void afterDump(@NonNull IConnection connection)
    {
        super.afterDump(connection);

        if (!(connection instanceof Connection))
        {
            throw new ParseException("Unsupported connection logType : " + connection.getClass().getSimpleName());
        }

        if (this.connection != null)
        {
            try
            {
                this.connection.disconnect();
            } catch (IOException e)
            {
                log.error("ERROR # disconnect metadata connection for address:{}", this.connection.getConnector()
                        .getSocketAddress(), e);
            }
        }
    }


    protected TimerTask buildHeartBeatTimeTask(IConnection connection)
    {
        if (!(connection instanceof Connection))
        {
            throw new ParseException("Unsupported connection logType : " + connection.getClass().getSimpleName());
        }

        // 开始mysql心跳sql
        if (detectingEnable && $.isNotNullAndNotEmpty(detectingSQL))
        {
            return new DetectingTimeTask((Connection) connection.fork());
        } else
        {
            return super.buildHeartBeatTimeTask(connection);
        }

    }

    protected void stopHeartBeatTimerTask()
    {
        TimerTask heartBeatTimerTask = this.heartBeatTimerTask;
        super.stopHeartBeatTimerTask();
        if (heartBeatTimerTask != null && heartBeatTimerTask instanceof DetectingTimeTask)
        {
            Connection connection = ((DetectingTimeTask) heartBeatTimerTask).getConnection();
            try
            {
                connection.disconnect();
            } catch (IOException e)
            {
                log.error("ERROR # disconnect heartbeat connection for address:{}", connection.getConnector()
                        .getSocketAddress(), e);
            }
        }
    }


    private Connection buildConnection(AuthenticationInfo authenticationInfo)
    {
        Connection connection = new Connection(//
                authenticationInfo.getInetSocketAddress(),//
                authenticationInfo.getDefaultDatabaseName(), authenticationInfo.getUsername(),//
                authenticationInfo.getPassword(),//
                connectionCharsetNumber//
        );//
        connection.getConnector().setReceiveBufferSize(receiveBufferSize);
        connection.getConnector().setSendBufferSize(sendBufferSize);
        connection.getConnector().setSoTimeout(defaultConnectionTimeoutInSeconds * 1000);
        connection.setCharset(connectionCharset);
        connection.setReceivedBinlogBytes(receivedBinlogBytes);

        connection.setSlaveId(generateUniqueServerId());//暂时这样
        return connection;
    }

    // =================== helper method =================

    private final long generateUniqueServerId()
    {
        try
        {
            // a=`echo $masterip|cut -d\. -f1`
            // b=`echo $masterip|cut -d\. -f2`
            // c=`echo $masterip|cut -d\. -f3`
            // d=`echo $masterip|cut -d\. -f4`
            // #server_id=`expr $a \* 256 \* 256 \* 256 + $b \* 256 \* 256 + $c
            // \* 256 + $d `
            // #server_id=$b$c$d
            // server_id=`expr $b \* 256 \* 256 + $c \* 256 + $d `
            InetAddress localHost = InetAddress.getLocalHost();
            byte[] address = localHost.getAddress();
            int salt = (destination != null) ? destination.hashCode() : 0;
            return ((0x7f & salt) << 24) + ((0xff & (int) address[1]) << 16) // NL
                    + ((0xff & (int) address[2]) << 8) // NL
                    + (0xff & (int) address[3]);
        } catch (UnknownHostException e)
        {
            throw new ParseException("Unknown host", e);
        }
    }

    protected EntryPosition findStartPosition(IConnection connection) throws IOException
    {
        if (isGTIDMode())
        {
            // GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instanc配置中的
            LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination);
            if (logPosition != null)
            {
                return logPosition.getEntryPosition();
            }

            if ($.isNotNullAndNotEmpty(entryPosition.getGtid()))
            {
                return entryPosition;
            }
        }

        EntryPosition startPosition = findStartPositionInternal(connection);
        if (needTransactionPosition.get())
        {
            log.warn("prepare to find last setIndexAsOriginIndexAndOffset$originIndex : {}", startPosition.toString());
            Long preTransactionStartPosition = findTransactionBeginPosition(connection, startPosition);
            if (!preTransactionStartPosition.equals(startPosition.getPosition()))
            {
                log.warn("find new start Transaction Position , old : {} , new : {}",
                        startPosition.getPosition(),
                        preTransactionStartPosition);
                startPosition.setPosition(preTransactionStartPosition);
            }
            needTransactionPosition.compareAndSet(true, false);
        }
        return startPosition;
    }


    protected EntryPosition findEndPositionWithMasterIdAndTimestamp(Connection connection)
    {

        final EntryPosition endPosition = findEndPosition(connection);
        if (tableMetaDataGetter != null)
        {
            long startTimestamp = System.currentTimeMillis();
            return findAsPerTimestampInSpecificLogFile(connection,
                    startTimestamp,
                    endPosition,
                    endPosition.getJournalName(),
                    true);
        } else
        {
            return endPosition;
        }
    }

    protected EntryPosition findPositionWithMasterIdAndTimestamp(Connection connection, EntryPosition fixedPosition)
    {

        if (tableMetaDataGetter != null && (fixedPosition.getTimestamp() == null || fixedPosition.getTimestamp() <= 0))
        {
            // 使用一个未来极大的时间，基于位点进行定位
            long startTimestamp = System.currentTimeMillis() + 102L * 365 * 24 * 3600 * 1000; // 当前时间的未来102年
            EntryPosition entryPosition = findAsPerTimestampInSpecificLogFile(connection,
                    startTimestamp,
                    fixedPosition,
                    fixedPosition.getJournalName(),
                    true);
            if (entryPosition == null)
            {
                throw new ParseException("[fixed timestamp] can't found begin/commit setIndexAsOriginIndexAndOffset$originIndex before with fixed setIndexAsOriginIndexAndOffset$originIndex"
                        + fixedPosition.getJournalName() + ":" + fixedPosition.getPosition());
            }
            return entryPosition;
        } else
        {
            return fixedPosition;
        }
    }

    protected EntryPosition findStartPositionInternal(IConnection connection)
    {
        Connection connection1 = (Connection) connection;
        LogPosition logPosition = logPositionManager.getLatestIndexBy(destination);
        if (logPosition == null)
        {// 找不到历史成功记录
            EntryPosition entryPosition = null;
            if (authenticationInfo != null && connection1.getConnector().getSocketAddress().equals(authenticationInfo.getInetSocketAddress()))
            {
                entryPosition = this.entryPosition;
            }

            if (entryPosition == null)
            {
                entryPosition = findEndPositionWithMasterIdAndTimestamp(connection1); // 默认从当前最后一个位置进行消费
            }

            // 判断一下是否需要按时间订阅
            if ($.isNullOrEmpty(entryPosition.getJournalName()))
            {
                // 如果没有指定binlogName，尝试按照timestamp进行查找
                if (entryPosition.getTimestamp() != null && entryPosition.getTimestamp() > 0L)
                {
                    log.warn("prepare to find start setIndexAsOriginIndexAndOffset$originIndex {}:{}:{}",
                            new Object[]{"", "", entryPosition.getTimestamp()});
                    return findByStartTimeStamp(connection1, entryPosition.getTimestamp());
                } else
                {
                    log.warn("prepare to find start setIndexAsOriginIndexAndOffset$originIndex just show master status");
                    return findEndPositionWithMasterIdAndTimestamp(connection1); // 默认从当前最后一个位置进行消费
                }
            } else
            {
                if (entryPosition.getPosition() != null && entryPosition.getPosition() > 0L)
                {
                    // 如果指定binlogName + offest，直接返回
                    entryPosition = findPositionWithMasterIdAndTimestamp(connection1, entryPosition);
                    log.warn("prepare to find start setIndexAsOriginIndexAndOffset$originIndex {}:{}:{}",
                            new Object[]{entryPosition.getJournalName(), entryPosition.getPosition(),
                                    entryPosition.getTimestamp()});
                    return entryPosition;
                } else
                {
                    EntryPosition specificLogFilePosition = null;
                    if (entryPosition.getTimestamp() != null && entryPosition.getTimestamp() > 0L)
                    {
                        // 如果指定binlogName +
                        // timestamp，但没有指定对应的offest，尝试根据时间找一下offest
                        EntryPosition endPosition = findEndPosition(connection1);
                        if (endPosition != null)
                        {
                            log.warn("prepare to find start setIndexAsOriginIndexAndOffset$originIndex {}:{}:{}",
                                    new Object[]{entryPosition.getJournalName(), "", entryPosition.getTimestamp()});
                            specificLogFilePosition = findAsPerTimestampInSpecificLogFile(connection1,
                                    entryPosition.getTimestamp(),
                                    endPosition,
                                    entryPosition.getJournalName(),
                                    true);
                        }
                    }

                    if (specificLogFilePosition == null)
                    {
                        // position不存在，从文件头开始
                        entryPosition.setPosition(BINLOG_START_OFFEST);
                        return entryPosition;
                    } else
                    {
                        return specificLogFilePosition;
                    }
                }
            }
        } else
        {
            if (logPosition.getLogIdentity().getSocketAddress().equals(connection1.getConnector().getSocketAddress()))
            {
                if (dumpErrorCountThreshold >= 0 && dumpErrorCount > dumpErrorCountThreshold)
                {
                    // binlog定位位点失败,可能有两个原因:
                    // 1. binlog位点被删除
                    // 2.vip模式的mysql,发生了主备切换,判断一下serverId是否变化,针对这种模式可以发起一次基于时间戳查找合适的binlog位点
                    boolean case2 = logPosition.getEntryPosition().getServerId() != null
                            && !logPosition.getEntryPosition().getServerId().equals(findServerId(connection1));
                    if (case2)
                    {
                        long timestamp = logPosition.getEntryPosition().getTimestamp();
                        long newStartTimestamp = timestamp - fallbackIntervalInSeconds * 1000;
                        log.warn("prepare to find start setIndexAsOriginIndexAndOffset$originIndex by last setIndexAsOriginIndexAndOffset$originIndex {}:{}:{}", new Object[]{"", "",
                                logPosition.getEntryPosition().getTimestamp()});
                        EntryPosition findPosition = findByStartTimeStamp(connection1, newStartTimestamp);
                        // 重新置为一下
                        dumpErrorCount = 0;
                        return findPosition;
                    }
                }
                // 其余情况
                log.warn("prepare to find start setIndexAsOriginIndexAndOffset$originIndex just last setIndexAsOriginIndexAndOffset$originIndex\n {}",
                        JsonUtils.marshalToString(logPosition));
                return logPosition.getEntryPosition();
            } else
            {
                // 针对切换的情况，考虑回退时间
                long newStartTimestamp = logPosition.getEntryPosition().getTimestamp() - fallbackIntervalInSeconds * 1000;
                log.warn("prepare to find start setIndexAsOriginIndexAndOffset$originIndex by switch {}:{}:{}", new Object[]{"", "",
                        logPosition.getEntryPosition().getTimestamp()});
                return findByStartTimeStamp(connection1, newStartTimestamp);
            }
        }
    }

    // 根据想要的position，可能这个position对应的记录为rowdata，需要找到事务头，避免丢数据
    // 主要考虑一个事务执行时间可能会几秒种，如果仅仅按照timestamp相同，则可能会丢失事务的前半部分数据
    private Long findTransactionBeginPosition(IConnection connection, final EntryPosition entryPosition)
            throws IOException
    {
        // 针对开始的第一条为非Begin记录，需要从该binlog扫描
        final java.util.concurrent.atomic.AtomicLong preTransactionStartPosition = new java.util.concurrent.atomic.AtomicLong(0L);
        connection.reconnect();
        connection.seek(entryPosition.getJournalName(), 4L, new SinkFunction<LogEvent>()
        {

            private LogPosition lastPosition;

            public boolean sink(LogEvent event)
            {
                try
                {
                    Entry entry = parse(event, true);
                    if (entry == null)
                    {
                        return true;
                    }

                    // 直接查询第一条业务数据，确认是否为事务Begin
                    // 记录一下transaction begin setIndexAsOriginIndexAndOffset$originIndex
                    if (entry.getEntryType() == EntryType.transaction_begin
                            && entry.getHeader().getLogFileOffset() < entryPosition.getPosition())
                    {
                        preTransactionStartPosition.set(entry.getHeader().getLogFileOffset());
                    }

                    if (entry.getHeader().getLogFileOffset() >= entryPosition.getPosition())
                    {
                        return false;// 退出
                    }

                    lastPosition = buildLastPosition(entry);
                } catch (Exception e)
                {
                    processSinkError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
                    return false;
                }

                return running;
            }
        });

        // 判断一下找到的最接近position的事务头的位置
        if (preTransactionStartPosition.get() > entryPosition.getPosition())
        {
            log.error("preTransactionEndPosition greater than startPosition from zk or localconf, maybe lost data");
            throw new ParseException("preTransactionStartPosition greater than startPosition from zk or localconf, maybe lost data");
        }
        return preTransactionStartPosition.get();
    }

    // 根据时间查找binlog位置
    private EntryPosition findByStartTimeStamp(Connection connection, Long startTimestamp)
    {
        EntryPosition endPosition = findEndPosition(connection);
        EntryPosition startPosition = findStartPosition(connection);
        String maxBinlogFileName = endPosition.getJournalName();
        String minBinlogFileName = startPosition.getJournalName();
        log.info("show master status to set search end condition:{} ", endPosition);
        String startSearchBinlogFile = endPosition.getJournalName();
        boolean shouldBreak = false;
        while (running && !shouldBreak)
        {
            try
            {
                EntryPosition entryPosition = findAsPerTimestampInSpecificLogFile(connection,
                        startTimestamp,
                        endPosition,
                        startSearchBinlogFile,
                        false);
                if (entryPosition == null)
                {
                    if (StringUtils.equalsIgnoreCase(minBinlogFileName, startSearchBinlogFile))
                    {
                        // 已经找到最早的一个binlog，没必要往前找了
                        shouldBreak = true;
                        log.warn("Didn't find the corresponding binlog files from {} to {}",
                                minBinlogFileName,
                                maxBinlogFileName);
                    } else
                    {
                        // 继续往前找
                        int binlogSeqNum = Integer.parseInt(startSearchBinlogFile.substring(startSearchBinlogFile.indexOf(".") + 1));
                        if (binlogSeqNum <= 1)
                        {
                            log.warn("Didn't find the corresponding binlog files");
                            shouldBreak = true;
                        } else
                        {
                            int nextBinlogSeqNum = binlogSeqNum - 1;
                            String binlogFileNamePrefix = startSearchBinlogFile.substring(0,
                                    startSearchBinlogFile.indexOf(".") + 1);
                            String binlogFileNameSuffix = String.format("%06d", nextBinlogSeqNum);
                            startSearchBinlogFile = binlogFileNamePrefix + binlogFileNameSuffix;
                        }
                    }
                } else
                {
                    log.info("found and return:{} in findByStartTimeStamp operation.", entryPosition);
                    return entryPosition;
                }
            } catch (Exception e)
            {
                log.warn(String.format("the binlogfile:%s doesn't exist, to continue to search the next binlogfile , caused by",
                        startSearchBinlogFile),
                        e);
                int binlogSeqNum = Integer.parseInt(startSearchBinlogFile.substring(startSearchBinlogFile.indexOf(".") + 1));
                if (binlogSeqNum <= 1)
                {
                    log.warn("Didn't find the corresponding binlog files");
                    shouldBreak = true;
                } else
                {
                    int nextBinlogSeqNum = binlogSeqNum - 1;
                    String binlogFileNamePrefix = startSearchBinlogFile.substring(0,
                            startSearchBinlogFile.indexOf(".") + 1);
                    String binlogFileNameSuffix = String.format("%06d", nextBinlogSeqNum);
                    startSearchBinlogFile = binlogFileNamePrefix + binlogFileNameSuffix;
                }
            }
        }
        // 找不到
        return null;
    }

    /**
     * 查询当前db的serverId信息
     */
    private Long findServerId(Connection connection)
    {
        try
        {
            ResultSetPacket resultSetPacket = connection.query("show variables like 'server_id'");
            List<String> fieldValueList = resultSetPacket.getFieldValueList();
            if (!$.isHasData(fieldValueList))
            {
                throw new ParseException("command : show variables like 'server_id' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
            }
            return Long.valueOf(fieldValueList.get(1));
        } catch (IOException e)
        {
            throw new ParseException("command : show variables like 'server_id' has an error!", e);
        }
    }

    /**
     * 查询当前的binlog位置
     */
    private EntryPosition findEndPosition(Connection connection)
    {
        try
        {
            ResultSetPacket resultSetPacket = connection.query("show master status");
            List<String> fieldValueList = resultSetPacket.getFieldValueList();
            if (!$.isHasData(fieldValueList))
            {
                throw new ParseException("command : 'show master status' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
            }
            EntryPosition endPosition = new EntryPosition(fieldValueList.get(0), Long.valueOf(fieldValueList.get(1)));
            if (isGTIDMode && fieldValueList.size() > 4)
            {
                endPosition.setGtid(fieldValueList.get(4));
            }
            return endPosition;
        } catch (IOException e)
        {
            throw new ParseException("command : 'show master status' has an error!", e);
        }
    }

    /**
     * 查询当前的binlog位置
     */
    private EntryPosition findStartPosition(Connection connection)
    {
        try
        {
            ResultSetPacket resultSetPacket = connection.query("show binlog events limit 1");
            List<String> fieldValueList = resultSetPacket.getFieldValueList();
            if (!$.isHasData(fieldValueList))
            {
                throw new ParseException("command : 'show binlog events limit 1' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
            }
            EntryPosition endPosition = new EntryPosition(fieldValueList.get(0), Long.valueOf(fieldValueList.get(1)));
            return endPosition;
        } catch (IOException e)
        {
            throw new ParseException("command : 'show binlog events limit 1' has an error!", e);
        }

    }

    /**
     * 查询当前的slave视图的binlog位置
     */
    @SuppressWarnings("unused")
    private SlaveEntryPosition findSlavePosition(Connection connection)
    {
        try
        {
            ResultSetPacket resultSetPacket = connection.query("show slave status");
            List<FieldPacket> names = resultSetPacket.getFieldDescriptorList();
            List<String> fields = resultSetPacket.getFieldValueList();
            if (!$.isHasData(fields))
            {
                return null;
            }

            int i = 0;
            Map<String, String> maps = new HashMap<String, String>(names.size(), 1f);
            for (FieldPacket name : names)
            {
                maps.put(name.getColumnName(), fields.get(i));
                i++;
            }

            String errno = maps.get("Last_Errno");
            String slaveIORunning = maps.get("Slave_IO_Running"); // Slave_SQL_Running
            String slaveSQLRunning = maps.get("Slave_SQL_Running"); // Slave_SQL_Running
            if ((!"0".equals(errno)) || (!"Yes".equalsIgnoreCase(slaveIORunning))
                    || (!"Yes".equalsIgnoreCase(slaveSQLRunning)))
            {
                log.warn("Ignoring failed slave: " + connection.getConnector().getSocketAddress() + ", Last_Errno = "
                        + errno + ", Slave_IO_Running = " + slaveIORunning + ", Slave_SQL_Running = "
                        + slaveSQLRunning);
                return null;
            }

            String masterHost = maps.get("Master_Host");
            String masterPort = maps.get("Master_Port");
            String binlog = maps.get("Master_Log_File");
            String position = maps.get("Exec_Master_Log_Pos");
            return new SlaveEntryPosition(binlog, Long.valueOf(position), masterHost, masterPort);
        } catch (IOException e)
        {
            log.error("find slave setIndexAsOriginIndexAndOffset$originIndex error", e);
        }

        return null;
    }

    /**
     * 根据给定的时间戳，在指定的binlog中找到最接近于该时间戳(必须是小于时间戳)的一个事务起始位置。
     * 针对最后一个binlog会给定endPosition，避免无尽的查询
     */
    private EntryPosition findAsPerTimestampInSpecificLogFile(Connection connection,
                                                              final Long startTimestamp,
                                                              final EntryPosition endPosition,
                                                              final String searchBinlogFile,
                                                              final Boolean justForPositionTimestamp)
    {

        final LogPosition logPosition = new LogPosition();
        try
        {
            connection.reconnect();
            // 开始遍历文件
            connection.seek(searchBinlogFile, 4L, new SinkFunction<LogEvent>()
            {

                private LogPosition lastPosition;

                public boolean sink(LogEvent logEvent)
                {
                    EntryPosition entryPosition = null;
                    try
                    {
                        Entry entry = parse(logEvent, true);
                        if (justForPositionTimestamp && logPosition.getEntryPosition() == null && logEvent.getCreateDateTimeLong() > 0)
                        {
                            // 初始位点
                            entryPosition = new EntryPosition(logEvent.getServerId(), searchBinlogFile,
                                    logEvent.getLogPosition(),
                                    logEvent.getCreateDateTimeLong() * 1000
                            );
                            logPosition.setEntryPosition(entryPosition);
                        }

                        if (entry == null)
                        {
                            return true;
                        }
                        Long serverId = entry.getHeader().getServerId();
                        //
                        String logFileName = entry.getHeader().getLogFileName();
                        Long logFileOffset = entry.getHeader().getLogFileOffset();
                        Long executeDateTimeLong = entry.getHeader().getExecuteDateTimeLong();


                        // 如果最小的一条记录都不满足条件，可直接退出
                        if (executeDateTimeLong >= startTimestamp)
                        {
                            return false;
                        }

                        if (StringUtils.equals(endPosition.getJournalName(), logFileName)
                                && endPosition.getPosition() <= logFileOffset)
                        {
                            return false;
                        }

                        // 记录一下上一个事务结束的位置，即下一个事务的position
                        // setIndexAsOriginIndexAndOffset$originIndex = current +
                        // data.length，代表该事务的下一条offest，避免多余的事务重复
                        if (EntryType.transaction_end.equals(entry.getEntryType()))
                        {
                            entryPosition = new EntryPosition(serverId, logFileName, logFileOffset, executeDateTimeLong);
                            logPosition.setEntryPosition(entryPosition);
                        } else if (EntryType.transaction_begin.equals(entry.getEntryType()))
                        {
                            // 当前事务开始位点
                            entryPosition = new EntryPosition(serverId, logFileName, logFileOffset, executeDateTimeLong);
                            logPosition.setEntryPosition(entryPosition);
                        }

                        lastPosition = buildLastPosition(entry);
                    } catch (Throwable e)
                    {
                        processSinkError(e, lastPosition, searchBinlogFile, 4L);
                    }

                    return running;
                }
            });

        } catch (IOException e)
        {
            log.error("ERROR ## findAsPerTimestampInSpecificLogFile has an error", e);
        }

        if (logPosition.getEntryPosition() != null)
        {
            return logPosition.getEntryPosition();
        } else
        {
            return null;
        }
    }

    @Override
    protected void processDumpError(Throwable e)
    {
        if (e instanceof IOException)
        {
            String message = e.getMessage();
            if (StringUtils.contains(message, "errno = 1236"))
            {
                // 1236 errorCode代表ER_MASTER_FATAL_ERROR_READING_BINLOG
                dumpErrorCount++;
            }
        }

        super.processDumpError(e);
    }


    /**
     * 心跳信息
     */
    class DetectingTimeTask extends TimerTask
    {

        @Setter
        @Getter
        private boolean reconnect = false;
        @Setter
        @Getter
        private Connection connection;

        public DetectingTimeTask(Connection connection)
        {
            this.connection = connection;
        }

        public void run()
        {
            try
            {
                if (reconnect)
                {
                    reconnect = false;
                    connection.reconnect();
                } else if (!connection.isConnected())
                {
                    connection.connect();
                }
                Long startTime = System.currentTimeMillis();

                // 可能心跳sql为select 1
                if (StringUtils.startsWithIgnoreCase(detectingSQL.trim(), "select")
                        || StringUtils.startsWithIgnoreCase(detectingSQL.trim(), "show")
                        || StringUtils.startsWithIgnoreCase(detectingSQL.trim(), "explain")
                        || StringUtils.startsWithIgnoreCase(detectingSQL.trim(), "desc"))
                {
                    connection.query(detectingSQL);
                } else
                {
                    connection.update(detectingSQL);
                }

                Long costTime = System.currentTimeMillis() - startTime;

            } catch (SocketTimeoutException e)
            {

                reconnect = true;
                log.warn("connect failed by ", e);
            } catch (IOException e)
            {

                reconnect = true;
                log.warn("connect failed by ", e);
            } catch (Throwable e)
            {

                reconnect = true;
                log.warn("connect failed by ", e);
            }

        }

    }

}
